/*
 * DataEngine.cpp
 *
 *  Created on: 2021��7��12��
 *      Author: root
 */

#include "DataEngine.h"

DataEngine::DataEngine()
{

    for (auto &it : eventBuffers)
    {
        it = make_unique<Disruptor<Event, DefaultRingBufferSize>>();
    }
}

DataEngine::~DataEngine()
{

    cout << "stop coreEngine" << endl;
    isStop = true;
}

int DataEngine::getCurrentToken()
{
    return m_MaxToken;
}

void DataEngine::PthreadBindCore(int core_id)
{
    cpu_set_t mask;
    CPU_ZERO(&mask);
    CPU_SET(core_id, &mask);
    pthread_setaffinity_np(pthread_self(), sizeof(mask), &mask);
}

void DataEngine::processQrySymbolStatusReport(EES_SymbolStatus *pSymbolStatus, bool bFinish)
{

    auto it = con_strategy_maps.find(pSymbolStatus->m_Symbol);
    if (it != con_strategy_maps.end())
    {
        set<BaseStrategy *> p_stra = it->second;

        for (const auto &p_strategy : p_stra)
        {
            p_strategy->preQryProcessSymbolStatus(pSymbolStatus->m_InstrumentStatus,pSymbolStatus->m_EnterTime);
        }
    }
}

void DataEngine::processSymbolStatusReport(EES_SymbolStatus *pSymbolStatus)
{

    Event event(pSymbolStatus);
    this->addEvent(event);
}

void DataEngine::resetCancelMap()
{
    auto it1 = this->informationMap.begin();
    while (it1 != this->informationMap.end())
    {
        it1->second = 0;
        ++it1;
    }
}

void DataEngine::resetPosition()
{
    auto it = strategies.begin();
    while (it != strategies.end())
    {
        for (int i = 0; i < it->second->contractCodes.size(); ++i)
        {
            it->second->positions[i].bkYdVol = it->second->positions[i].bkYdVol + it->second->positions[i].bkTdVol;
            it->second->positions[i].skYdVol = it->second->positions[i].skYdVol + it->second->positions[i].skTdVol;
            it->second->positions[i].bkTdVol = 0;
            it->second->positions[i].skTdVol = 0;
        }
        //it->second->sumProfit=0;
        it->second->dealVol=0;
        ++it;
    }
}

void DataEngine::startAllStrategy()
{
    for (auto it = this->strategies.begin(); it != strategies.end(); ++it)
    {
        it->second->start();
        usleep(1000 * 100);
    }
}

void DataEngine::stopAllStrategy()
{
    for (auto it = this->strategies.begin(); it != strategies.end(); ++it)
    {
        it->second->stop();
        usleep(100 * 100);
    }
}
int DataEngine::getNextToken()
{
    // lock_guard<mutex> lock(mtx);
    return ++m_MaxToken;
}

/*void DataEngine::orderCancel(long long int Ees_orderId) {

    orderBuffer.WriteInBuf(ReqOrderField(Ees_orderId));
}*/


/*
void DataEngine::reqOrderInsert(ReqOrderField &orderField) {
    //orderRefMaps.insert(orderField.enterOrder.m_ClientOrderToken,contractBufferMap[orderField.enterOrder.m_Symbol]);
    orderBuffer.WriteInBuf(orderField);
}
*/

void DataEngine::udpOrderHandel(ReqOrderMsg &orderMsg)
{
    //lock_guard<mutex> lck(mtx);
    udpBuffer.WriteInBuf(orderMsg);
}
/*EES_ClientToken
DataEngine::orderInsert(char exchangeId, const string &contractCode, double price, unsigned int vol, char flag,
                        BaseStrategy *p_strategy) {

    lock_guard<mutex> lock(mtx);

    EES_ClientToken token = ++m_MaxToken;
    ReqOrderField temp;
    strcpy(temp.enterOrder.m_Symbol, contractCode.c_str());
    temp.enterOrder.m_Side = flag;
    temp.enterOrder.m_Exchange = exchangeId;
    temp.enterOrder.m_Price = price;
    temp.enterOrder.m_Qty = vol;
    temp.enterOrder.m_ForceMarketSessionId = true;
    temp.enterOrder.m_ClientOrderToken = token;
    temp.type = INSERT;
    orderRefMaps.insert(
            pair<EES_ClientToken, Disruptor<Event, DefaultRingBufferSize> *>(token,
                                                                             contractBufferMap[contractCode]));
    orderBuffer.WriteInBuf(temp);
    return token;


}*/

void DataEngine::saveCancelMap()
{
    RedisImp r("127.0.0.1");

    auto it = this->informationMap.begin();
    while (it != informationMap.end())
    {
        r.getRedis().hset("informationVols", it->first, std::to_string(it->second));
        ++it;
    }
    r.getRedis().set("maxStrategyId", std::to_string(maxStrategyId));
    for(const auto& item:enterTime_map)
    {
        r.getRedis().hset("notice",item.first,item.second.toJson());
    }
}

void DataEngine::loadCancelMap()
{
    map<string,string> informationTimes;
    map<string,string> tmpOffset;
    RedisImp r("127.0.0.1");
    r.getRedis().hgetall("informationVols",std::inserter(informationTimes,informationTimes.begin()));
    auto it = informationTimes.begin();
    while (it != informationTimes.end())
    {
        this->informationMap[it->first] = stoi(it->second);
        ++it;
    }
    if(r.getRedis().exists("maxStrategyId"))
    {
        maxStrategyId=stoi(r.getRedis().get("maxStrategyId").value());
        maxStrategyId++;
    }
    if(r.getRedis().exists("notice"))
    {
        r.getRedis().hgetall("notice",std::inserter(tmpOffset,tmpOffset.begin()));
        for(const auto &item:tmpOffset)
        {
            enterTime_map[item.first].fromJson(item.second);
        }
    }
}
void DataEngine::loadContracts() {
    RedisImp r("150.158.44.12");
    map<string,string> tmpMaps;
    r.getRedis().hgetall("AllContract",std::inserter(tmpMaps,tmpMaps.begin()));
    for(const auto &item:tmpMaps)
    {
        auto j=json::parse(item.second);
        if(j["exchange_id"]==103||j["exchange_id"]==106)
        {
            contractMap.insert(make_pair(item.first,j));
        }
    }
}
long DataEngine::getNanoTime()
{
    auto now = std::chrono::system_clock::now();
    // 将时间戳转换为毫秒数
    auto now_us = std::chrono::time_point_cast<std::chrono::microseconds>(now);
    return now_us.time_since_epoch().count();
}

long DataEngine::calculateTime(int hour,int minute,int second)
{
    auto now = std::chrono::system_clock::now();
    auto now_c = std::chrono::system_clock::to_time_t(now);
    std::tm* now_tm = std::localtime(&now_c);
    // 目标时间，下午1点30分0秒
    std::tm target = *now_tm;
    target.tm_hour = hour;
    target.tm_min = minute;
    target.tm_sec = second;
    target.tm_isdst = -1; // 让系统决定夏令时
    // 转换为time_t
    auto target_time_t = std::mktime(&target);
    // 计算时间差
    auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::from_time_t(target_time_t) - now);
    return duration.count();
}
void DataEngine::delStrategy(BaseStrategy *p_strategy) {
    if (p_strategy != nullptr) {
        p_strategy->stop();
        this->strategies.erase(p_strategy->strategyId);
        this->strategyMap.erase(p_strategy->strategyName);

        // 使用迭代器的 erase 方法来避免手动 ++ 导致的迭代器失效问题
        for (auto it = p_strategy->contractCodes.begin(); it != p_strategy->contractCodes.end(); ) {
            const string& contractCode = *it;
            auto its = con_strategy_maps.find(contractCode);
            if (its != con_strategy_maps.end()) {
                its->second.erase(p_strategy);
            }
            it = p_strategy->contractCodes.erase(it);  // 使用 erase 返回的新迭代器
        }

        RedisImp redisImp("127.0.0.1");
        try {
            if (redisImp.getRedis().hexists("strategy", p_strategy->strategyName)) {
                redisImp.getRedis().hdel("strategy", p_strategy->strategyName);
            }
            if (redisImp.getRedis().hexists("position", p_strategy->strategyName)) {
                redisImp.getRedis().hdel("position", p_strategy->strategyName);
            }
        } catch (const std::exception& e) {
            // 异常处理
            std::cerr << "Redis operation failed: " << e.what() << std::endl;
        }

        this->initEngine();
        delete p_strategy;
    } else {
        // 处理 p_strategy 为 nullptr 的情况
        std::cerr << "p_strategy is nullptr" << std::endl;
    }
}

void DataEngine::addStrategy(int index, BaseStrategy *p_strategy)
{
    this->strategies[index] = p_strategy;
    this->strategyMap[p_strategy->strategyName]=p_strategy;
    auto it = p_strategy->contractCodes.begin();
    while (it != p_strategy->contractCodes.end())
    {
        string contractCode = (*it);

        auto its = con_strategy_maps.find(contractCode);
        if (its != this->con_strategy_maps.end())
        {
            its->second.insert(p_strategy);
        }
        else
        {
            set<BaseStrategy *> sets;
            sets.insert(p_strategy);
            con_strategy_maps.insert(pair<string, set<BaseStrategy *>>(contractCode, sets));

            contractBufferMap.insert(make_pair(contractCode, eventBuffers[getIndexOfContract(contractCode)].get()));
        }
        ++it;
    }
}

int DataEngine::getIndexOfContract(const std::string &s)
{
    if (s.empty()) {
        throw std::invalid_argument("Input string is too short");
    }
    int tmp = int(s[0]) + 3 - 100;
    return tmp / 7;
}

void DataEngine::addEvent(Event &event)
{

    if (event.type == TICK)
    {
        contractBufferMap[event.tick.instrument_id]->WriteInBuf(event);
    }
    else if (event.type == ORDER)
    {
        contractBufferMap[event.order.contractCode]->WriteInBuf(event);
    }
    else
    {
        for (const auto &eventBuffer : this->eventBuffers)
        {
            eventBuffer->WriteInBuf(event);
        }
    }
}

void DataEngine::startEngine()
{
    for (int i = 0; i < size(eventBuffers); i++)
    {
        Disruptor<Event, DefaultRingBufferSize> *p_eventBuffer = eventBuffers[i].get();
        thread t([p_eventBuffer, i, this]
                 {
            cpu_set_t mask;
            CPU_ZERO(&mask);
            CPU_SET(i + 2, &mask);
            pthread_setaffinity_np(pthread_self(), sizeof(mask), &mask);
            while (true) {
                if (p_eventBuffer->empty()) {
                    if (isStop) {
                        break;
                    }
                } else {
                    BufConsumer<Event> my_bufConsumer(p_eventBuffer);
                    Event &event = my_bufConsumer.GetContent();
                    if (event.type == ORDER) {

                        auto it = bufferStrategyMap.find(p_eventBuffer);
                        if (it != bufferStrategyMap.end()) {
                            for (const auto &item: bufferStrategyMap[p_eventBuffer]) {
                                item->preProcessOrder(event.order);
                            }
                        }

                    } else if (event.type == TICK) {
                        event.tick.fmtTime();
                        for (const auto &p_str: con_strategy_maps[event.tick.instrument_id]) {
                            p_str->preProcessMD(event.tick);
                        }

                    }
                    else
                    {
                        auto it = bufferStrategyMap.find(p_eventBuffer);
                        if (it != bufferStrategyMap.end()) {
                            for (const auto &item: bufferStrategyMap[p_eventBuffer]) {
                                if(event.m_InstrumentStatus == '2')
                                {
                                    isTrading= true;
                                }
                                item->preProcessSymbolStatus(event.m_InstrumentStatus,event.m_EnterTime);
                            }
                        }
                        SPDLOG_INFO("{} status =  {}",event.m_EnterTime,int(event.m_InstrumentStatus));
                    }
                }
            } });
        t.detach();
    }
}

int DataEngine::getNextStrategyId() {
    return this->strategyId++;
}
void DataEngine::initEngine()
{
    bufferStrategyMap.clear();
    for (const auto &item : contractBufferMap)
    {
        string contractCode = item.first;
        auto it = con_strategy_maps.find(contractCode);
        for (const auto &itStrategy : it->second)
        {
            bufferStrategyMap[item.second].insert(itStrategy);
        }
    }
}

DataEngine *DataEngine::getInstancePtr()
{
    static DataEngine instance; // 局部静态变量，确保只初始化一次
    return &instance;           // 返回单例的地址
}

bool DataEngine::mdIsEffective(const TickDataHf &md0, const TickDataHf &md1) {
    int64_t timeDiff = (md0.update_time - md1.update_time);
    if(abs(timeDiff)<60)
    {
        int32_t milliSecDiff = (md0.update_milli_sec - md1.update_milli_sec);
        // 避免整数溢出
        return (timeDiff > 0) || (timeDiff == 0 && milliSecDiff >= 0);
    }
    return false;
}
double DataEngine::formatDouble(double price) {
    double roundedPrice = std::round(price * DECIMAL_FACTOR) / DECIMAL_FACTOR;
    return roundedPrice;
}

string DataEngine::getMaxStrategyId() {
    maxStrategyId++;
    return to_string(maxStrategyId);
}
